package org.tksj.bim.mq.service;

import io.netty.handler.codec.mqtt.MqttQoS;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.buffer.Buffer;
import io.vertx.mqtt.MqttClient;
import io.vertx.mqtt.MqttClientOptions;
import org.springframework.stereotype.Component;

@Component
public class MqVerticle  extends AbstractVerticle {


    public static  final  String Topic="cars/bim/test/#";
    MqttClient client=null;

    @Override
    public void start() {
        MqttClientOptions option=new MqttClientOptions();
        option.setClientId("client_plugin");
        option.setMaxMessageSize(100_000_000);
        option.setCleanSession(true);
        option.setAutoKeepAlive(true);
        client= MqttClient.create(vertx,option);
        client.connect(1883,"54.244.173.190",s->{
            if(s.succeeded()){
                System.out.println("连接成功");
                subscribe();
                client.publishHandler(handler->{
                    System.out.println("收到消息:"+handler.payload());
                });
            }
        });
    }

    private  void  publish(){

        Buffer payload=Buffer.buffer("client:ok");

        client.publish(Topic,payload, MqttQoS.AT_MOST_ONCE,false,false,handler->{
            if(handler.succeeded()){
                System.out.println("publish success");
            }
            else{
                System.out.println("publish failed");
            }
        });

    }
    private void  subscribe(){
       client.subscribe(Topic,0,e->{
          if(e.succeeded()){
              System.out.println("subscribe success");
          }
          else {
              System.out.println("subscribe failed");
          }
       });
    }

    @Override
    public void stop() {
        if(client!=null){
            if(client.isConnected()){
                client.disconnect();
            }
        }
    }


}
